package com.share.rabbitmq.quickstart;

import com.rabbitmq.client.*;
import com.share.rabbitmq.RabbitMQConnection;

import java.io.IOException;

/**
 * author:caifan
 * date:2020/3/1
 */
public class QuickConsumer {
    private static final String queueName = "queue_01";
    public static void main(String[] args) throws Exception {
        Connection connect = RabbitMQConnection.connect();
//        simpleConsumer(connect);
        messageAck(connect);

    }

    /**
     * 消息确认
     * @param connect
     */
    public static void messageAck(Connection connect) throws Exception {
        final String exchangeName = "my.exchange.direct";
        final String queueName = "my.exchange.queue";
        final String routingKey = "my.test.ack";
        Channel channel = connect.createChannel();
        channel.exchangeDeclare(exchangeName, "direct", true);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);

        channel.basicConsume(queueName, false, new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                System.out.println("receive message:{}" + new String(message.getBody()) + ":" + message.getEnvelope().getDeliveryTag() + "**" + consumerTag);
                int i = 1 / 0;
                channel.basicAck(message.getEnvelope().getDeliveryTag(), true);
            }
        }, new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {

            }
        });

    }

    /**
     * 简单的消费消息
     * @param connect
     * @throws Exception
     */
    public static void simpleConsumer(Connection connect) throws Exception {
        Channel channel = connect.createChannel();
        channel.queueDeclare(queueName, false, false, false, null);

        /*
        //默认消费者方式消费消息
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consuserTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println(envelope.getExchange() + ":" + envelope.getRoutingKey() + ":" + message);
            }
        };
        channel.basicConsume(queueName, true, consumer);
        */

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("consumer message:" + message);
        };
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
    }
}
